package com.sl.plugin.step.db;

import cn.hutool.core.util.StrUtil;
import com.bazaarvoice.jolt.Chainr;
import com.sl.core.utils.JsonUtil;
import com.sl.core.engine.meta.impl.AbstractProcessCompDescriptor;
import com.sl.core.engine.step.ProcessCompService;
import com.sl.core.engine.step.impl.AbstractProcessComp;
import com.sl.core.engine.utils.BeanUtil;
import lombok.SneakyThrows;
import org.anyline.adapter.KeyAdapter;
import org.anyline.entity.DataRow;
import org.anyline.metadata.Table;
import org.anyline.service.AnylineService;
import org.apache.commons.compress.utils.Lists;

import java.util.Collections;
import java.util.List;
import java.util.Map;

/**
 *
 */
public class DbInsertComp extends AbstractProcessComp<List<Map<String, Object>>, List<Map<String, Object>>> implements ProcessCompService {


    public DbInsertComp(AbstractProcessCompDescriptor abstractProcessCompDescriptor) {
        super(abstractProcessCompDescriptor);
    }

    @SneakyThrows
    @Override
    public List<Map<String, Object>> input(List<Map<String, Object>> param) {
        String jolt = this.meta.getStrAttr("jolt");
        if (StrUtil.isBlank(jolt)) {
            //未配置映射关系，直接返回null
            return Collections.emptyList();
        }
        List chainrSpecJSON = JsonUtil.toList(jolt, Map.class);
        Chainr chainr = Chainr.fromSpec(chainrSpecJSON);
        List<Map<String, Object>> result = Lists.newArrayList();
        for (Map<String, Object> map : param) {
            Object transformedOutput = chainr.transform(map);

            Map<String, Object> paramMap = BeanUtil.convert(transformedOutput);
            insertData(paramMap);
            result.add(paramMap);
            //开始插入
        }
        return result;
    }


    private void insertData(Map<String, Object> paramMap) {
        String dataSourceId = BeanUtil.getStr(meta.getAttributeMap(), "output.dataSourceId");
        String schemaId = BeanUtil.getStr(meta.getAttributeMap(), "output.schemaId");
        String tableName = BeanUtil.getStr(meta.getAttributeMap(), "output.tableName");
        //现在开始新增
//        DataSourceManager.INSTANCE.openTranscation(dataSourceId);

        AnylineService transactionAnyLineService = getAnyLineService(dataSourceId,false);
//        AnylineService anylineService = DataSourceManager.INSTANCE.anylineService(dataSourceId);
        DataRow dataRow = new DataRow(KeyAdapter.KEY_CASE.CAMEL_LOWER);
        dataRow.putAll(paramMap);

        Table table = new Table(tableName);
        table.setCatalog(schemaId);
        long insert = transactionAnyLineService.insert(table, dataRow);
        paramMap.put("rows", insert);
    }

}
